/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: DoubleAndTripleBoltApp
 * Author:   An-Il
 * Date:     2018/11/9 9:19
 * Description: wordcount实例
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.integration.jdbc;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.StormTopology;
import org.apache.storm.jdbc.common.Column;
import org.apache.storm.jdbc.common.ConnectionProvider;
import org.apache.storm.jdbc.common.HikariCPConnectionProvider;
import org.apache.storm.jdbc.mapper.JdbcMapper;
import org.apache.storm.jdbc.mapper.SimpleJdbcLookupMapper;
import org.apache.storm.jdbc.mapper.SimpleJdbcMapper;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.sql.Types;
import java.util.*;

/**
 * 使用storm实现wordcount实例,mysql存储
 *
 * @url http://storm.apache.org/releases/1.2.2/storm-redis.html
 */
public class LocalWordCountJdbcApp {

    private static String tableName = "wordcount";

    private static final String SELECT_QUERY = "select * from " + tableName + " where word = ?";

    /**
     * 数据源产生的Spout
     */
    public static class RandomWordSpout extends BaseRichSpout {

        SpoutOutputCollector collector;
        Random random;

        /**
         * 初始化
         *
         * @param conf      配置参数
         * @param context   上下文
         * @param collector 数据发射器
         */
        @Override
        public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
            this.collector = collector;
            this.random = new Random();
        }

        /**
         * 此方法是个死循环
         */
        @Override
        public void nextTuple() {
            Utils.sleep(1000);
            String[] words = new String[]{"hadoop", "spark", "kafka", "flume", "storm", "hbase", "hive", "zookeeper"};
            String word = words[random.nextInt(words.length)];

            //将数据发送到下游
            this.collector.emit(new Values(word), UUID.randomUUID());
            System.out.println("send val:" + word);
        }

        /**
         * 申明输出字段
         *
         * @param declarer
         */
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("word"));
        }

        @Override
        public void fail(Object msgId) {
            System.err.println(msgId + "处理失败");
        }
    }

    /**
     * 接受数据并处理
     */
    public static class WordCountBolt extends BaseRichBolt {

        OutputCollector collector;

        @Override
        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
            this.collector = collector;
        }

        @Override
        public void execute(Tuple tuple) {
            try {
                /**
                 * 这里通过上面定义的Fields进行获取
                 * 也可以通过下标获取：tuple.getInteger(0)
                 */
                String word = tuple.getStringByField("word");

                //输出
                this.collector.emit(tuple, new Values(1, word));
                this.collector.ack(tuple);
            } catch (Exception e) {
                this.collector.fail(tuple);
            }
        }

        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("newCount", "word"));
        }
    }


    public static class WordCountStoreMapper {

    }

    /**
     * 构建本地Topology
     *
     * @param args
     * @url http://storm.apache.org/releases/1.2.2/Local-mode.html
     * http://storm.apache.org/releases/1.2.2/Running-topologies-on-a-production-cluster.html
     */
    public static void main(String[] args) {

        /**
         * 定义拓扑,在生产群集上运行拓扑与在本地模式下运行类似
         * Topology需要指定相关的Spout和Bolt的执行顺序
         * shuffleGrouping("dataSourceSpout"):Tuples以一种随机分布方式在Bolt的任务中,每个Bolt都保证获得相同数量的Tuples
         */
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("RandomWordSpout", new RandomWordSpout());
        builder.setBolt("WordCountBolt", new WordCountBolt()).shuffleGrouping("RandomWordSpout");

        //存储到jdbc的bolt
        Map hikariConfigMap = Maps.newHashMap();
        hikariConfigMap.put("dataSourceClassName", "com.mysql.jdbc.jdbc2.optional.MysqlDataSource");
        hikariConfigMap.put("dataSource.url", "jdbc:mysql://localhost/shop_2018");
        hikariConfigMap.put("dataSource.user", "root");
        hikariConfigMap.put("dataSource.password", "root");
        ConnectionProvider connectionProvider = new HikariCPConnectionProvider(hikariConfigMap);

        /**
         * 1.数据库存在此word记录（整合的是官方的查询模块）
         * 2.判断数据库是否有记录？
         *  无记录：插入数据：整合的是官方提供的插入模块（这里必须使用自定义的sql，因为上游bolt输出的字段名和数据库不一样）
         *  有记录：将数据库记录值 + 1 进行更新
         */

        //1.判断(查询)：这里需要指定输出字段和查询条件
        Fields outputFields = new Fields("word", "count");
        List<Column> queryParamColumns = Lists.newArrayList(new Column("word", Types.VARCHAR));
        SimpleJdbcLookupMapper simpleJdbcLookupMapper = new SimpleJdbcLookupMapper(outputFields, queryParamColumns);

        //2.无记录：插入数据库，这里需要指定列名，从上游的bolt的获取值（注意字段名）
        List<Column> schemaColumns = Lists.newArrayList(new Column("word", Types.VARCHAR), new Column("newCount", Types.INTEGER));
        JdbcMapper jdbcInsertMapper = new SimpleJdbcMapper(schemaColumns);


        //2.有记录：更新数据库，这里需要指定列名和更新条件


        JdbcInsertOrUpdateBolt JdbcInsertOrUpdateBolt = new JdbcInsertOrUpdateBolt(connectionProvider, SELECT_QUERY, simpleJdbcLookupMapper)
                .withInsertQueryAndInsertMapper("insert into " + tableName + " (word, count) values (?,?)", jdbcInsertMapper)
                .withUpdateQuery("update " + tableName + " set count = ? where word = ?");

        builder.setBolt("JdbcInsertOrUpdateBolt", JdbcInsertOrUpdateBolt).shuffleGrouping("WordCountBolt");


        StormTopology topology = builder.createTopology();
        //本地模式模拟正在进行的Storm集群，只需使用LocalCluster类
        LocalCluster cluster = new LocalCluster();
        /**
         * 然后，您可以使用submitTopology该LocalCluster对象上的方法提交拓扑。就像StormSubmitter上的相应方法一样，
         * submitTopology采用名称，拓扑配置和拓扑对象。然后，您可以使用killTopology将拓扑名称作为参数的方法来终止拓扑。
         */
        cluster.submitTopology("LocalWordCountJdbcApp", new Config(), topology);

    }

}